package com.hz.cheetah.server;

import com.hz.cheetah.server.imp.RpcSession;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.socket.SocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.swing.*;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by yangchuan on 16/7/7.
 */
public class SessionsManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(SessionsManager.class);
    public static ConcurrentHashMap<Channel,RpcSession> sessions = new ConcurrentHashMap();

    private static final ChannelFutureListener remover = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            sessions.remove(future.channel());
        }
    };

    public  static RpcSession putIfAbsent(Channel channel,RpcSession session){
            return sessions.putIfAbsent(channel,session);
    }

    public static RpcSession resetSession(Channel channel){
        RpcSession session = sessions.get(channel);

        if(session == null){
            synchronized (channel){
                if(session == null){
                    session = new RpcSession((SocketChannel)channel);
                    if(sessions.putIfAbsent(channel,session) == null) {
                        channel.closeFuture().removeListener(remover);
                    }
                }
            }
        }
        session.reset();
        return session;
    }

    /**
     * 用于服务器之间有其它网络设备的服务器环境，如果服务器之间是直连的就不需要heartbeat
     * 防止出现大量close_waited
     */
    private static class watchdog implements Runnable {

        public void run() {
            Set<Map.Entry<Channel,RpcSession>> entrys = null;
            Iterator<Map.Entry<Channel,RpcSession>> iterator = null;
            Map.Entry<Channel,RpcSession> entry = null;
            RpcSession session = null;

            while (true) {
                try {
                    entrys = sessions.entrySet();
                    iterator = entrys.iterator();
                    while(iterator.hasNext()){
                        entry = iterator.next();
                        session = entry.getValue();
                        if (session == null) {
                            continue;
                        }
                        /**
                         * 如果10分钟（3 * 200 * 1000）没有收到对应客户端的链接，就主动强制关闭客户端的链接
                         */
                        if (System.currentTimeMillis() - session.lasttime >  200 * 1000) {
                            if(session.tryitCount < 2) {
                                session.addCount();
                                LOGGER.warn(" no get heatbeat msg:{},session id is:{}", session.tryitCount,session.getClientAddress());
                            }else{
                                session.close(true);
                                sessions.remove(entry.getKey());
                                session = null;
                            }
                        }
                    }
                    Thread.sleep(3000);
                    LOGGER.info("{},{}",sessions.size(),sessions);
                } catch (Throwable e) {
                    LOGGER.error("Exception when scan the client do disconnect.", e);
                }
            }
        }
    }

    static {
        Thread th = new Thread(new watchdog(), "ClientConnectScanTimer");
        th.setDaemon(true);
        th.start();
    }


}
